package com.atguigu.actual.edu0417.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.beans.TrafficHomeDetailPageViewBean;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.DorisSinkUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author: techyang
 * @since: 2023-10-21 09:59
 * @description: 首页,详情页页面浏览统计
 **/
public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        DwsTrafficHomeDetailPageViewWindow dwsTrafficHomeDetailPageView = new DwsTrafficHomeDetailPageViewWindow();
        dwsTrafficHomeDetailPageView.base(args);
    }

    @Override
    public void process(StreamExecutionEnvironment env) {
        // 读取页面日志数据
        DataStreamSource<String> kafkaStrDs = env.fromSource(MyKafkaUtil.getKafkaSource(EduConfig.KAFKA_DWD_TRAFFIC_PAGE_LOG, "dws_traffic_home_detail_page_view_window_group"),
                WatermarkStrategy.noWatermarks(), "kafka_source");

        // 转换为jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaStrDs.map(JSON::parseObject);

        // 过滤出首页和商品详情页的日志
        // {"common":{"sc":"1","ar":"15","uid":"868","os":"Android 11.0","ch":"oppo","is_new":"0","md":"vivo iqoo3","mid":"mid_209","vc":"v2.1.134","ba":"vivo","sid":"f3833cbc-2f0b-4a57-885d-41a8e1d4afc0"},"page":{"page_id":"home","during_time":8105},"ts":1697790514300}
        // {"common":{"sc":"3","ar":"25","uid":"765","os":"iOS 13.3.1","ch":"Appstore","is_new":"1","md":"iPhone X","mid":"mid_210","vc":"v2.1.132","ba":"iPhone","sid":"3cca8171-b398-49cc-9765-9dcc4bcc43f0"},"page":{"page_id":"course_list","item":"flink","during_time":17143,"item_type":"keyword","last_page_id":"course_detail"},"ts":1697790514314}
        SingleOutputStreamOperator<JSONObject> filterDs =
                jsonObjDs.filter(jsonObj -> ("home".equals(jsonObj.getJSONObject("page").getString("page_id"))
                        || "course_detail".equals(jsonObj.getJSONObject("page").getString("page_id"))
                && StringUtils.isNotEmpty(jsonObj.getJSONObject("common").getString("uid"))));

        // 分配时间戳和水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkDs = filterDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                (jsonObj, ts) -> jsonObj.getLong("ts")
                        )
        );

        // 转换为wordcount
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> beanDs = withWatermarkDs.map(new MapFunction<JSONObject, TrafficHomeDetailPageViewBean>() {
            @Override
            public TrafficHomeDetailPageViewBean map(JSONObject jsonObj) throws Exception {
                TrafficHomeDetailPageViewBean trafficHomeDetailPageViewBean = new TrafficHomeDetailPageViewBean();
                JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                String pageId = pageJsonObj.getString("page_id");
                if ("home".equals(pageId)) {
                    trafficHomeDetailPageViewBean.setHomePuCt(1L);
                    trafficHomeDetailPageViewBean.setCourseDetailPuCt(0L);
                } else {
                    trafficHomeDetailPageViewBean.setHomePuCt(0L);
                    trafficHomeDetailPageViewBean.setCourseDetailPuCt(1L);
                }
                trafficHomeDetailPageViewBean.setUserId(commonJsonObj.getString("uid"));
                return trafficHomeDetailPageViewBean;
            }
        });


        // 按照用户分组
        KeyedStream<TrafficHomeDetailPageViewBean, String> keyedByUserIdDs = beanDs.keyBy(TrafficHomeDetailPageViewBean::getUserId);

        // 开窗
        WindowedStream<TrafficHomeDetailPageViewBean, String, TimeWindow> windowDs = keyedByUserIdDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        // 聚合
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reduceDs =
                windowDs.reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                   @Override
                   public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {
                       value1.setHomePuCt(value1.getHomePuCt() + value2.getHomePuCt());
                       value1.setCourseDetailPuCt(value1.getCourseDetailPuCt() + value2.getCourseDetailPuCt());
                       return value1;
                   }
               },
                new WindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TrafficHomeDetailPageViewBean> input, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        String stt = DateFormatUtil.toYmdHms(start);
                        String edt = DateFormatUtil.toYmdHms(end);
                        String curDate = DateFormatUtil.toDate(start);
                        for (TrafficHomeDetailPageViewBean bean : input) {
                            bean.setStt(stt);
                            bean.setEdt(edt);
                            bean.setCurDate(curDate);
                            out.collect(bean);
                        }
                    }
                });
        reduceDs.print("reduceDs");
        reduceDs.addSink(DorisSinkUtil.<TrafficHomeDetailPageViewBean>getJdbcSink("insert into dws_traffic_home_detail_page_view_window values (?,?,?,?,?)"));

    }
}
